﻿#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#include <sys/epoll.h>
#include <sys/eventfd.h>
#include "SLT_log.h"

#include <sys/types.h>
#include <sys/socket.h>
#include <errno.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <netinet/tcp.h>

#include "event2/util.h"
#include <functional>

#include "SL_tcpClient.h"
#include "SL_tcpClient_impl.h"
#include "SL_tcpServer_epoll_impl.h"

#define _TCPSERVER_LOG_DEBUG(_format,...)   SLT_LOG(1,SLT_DEBUG,"[tcp]" _format, ##__VA_ARGS__)
#define _TCPSERVER_LOG_INFO(_format,...) SLT_LOG(1,SLT_INFO,"[tcp]" _format, ##__VA_ARGS__)
#define _TCPSERVER_LOG_ERR(_format,...) SLT_LOG(1,SLT_ERROR,"[tcp]" _format, ##__VA_ARGS__)
SLTcpServerEpollImpl::SLTcpServerEpollImpl()
{
}
int SLTcpServerEpollImpl::init(SLTcpServerCB* pcb)
{
	m_QuitEventfd = eventfd(0, 0);
	m_pcb = pcb;

	SLTcpTimeoutCheckSession* pcheckSession = new SLTcpTimeoutCheckSession();
	pcheckSession->m_ptcpServer = this;
	SLF_SessionBasePtr baseSee = std::shared_ptr<SLTcpTimeoutCheckSession>(pcheckSession);
	SLF_addSession(baseSee);
	return 0;
}
void SLTcpServerEpollImpl::start()
{
	m_thread = std::thread(std::bind(&SLTcpServerEpollImpl::thread_listen, this));
}
int SLTcpServerEpollImpl::addListen(std::string _ip, int _port)
{
	
	int sock = NetMakeSocket(_port);
	if (0 > sock) {
		_TCPSERVER_LOG_ERR("add listen:%s,%d,sock:%d,errorn:%d", _ip.c_str(), _port, sock,errno);
		return -1;
	}

	_TCPSERVER_LOG_INFO("add listen:%s,%d", _ip.c_str(), _port);
	//SLTcpClient::NoBlock(sock);
	evutil_make_socket_nonblocking(sock);

	//SendEventSig(sock, EPOLL_ADD_LISTEN_SOCKET);
	_tcp_event_info einfo;
	einfo.m_cmd = EPOLL_LIST_CMD_ADD_LISTEN_SOCKET;
	einfo.m_socket = sock;
	einfo.m_port = _port;
	postEvent(einfo);
	return 0;
}

int SLTcpServerEpollImpl::NetMakeSocket(int port)
{
	int nSocket, nValue = 1;
	if (-1 == (nSocket = socket(AF_INET, SOCK_STREAM, 0)))
	{
		//NetError(errStr, "Error on create socket : ", errno);
		return -1;
	}

	/* 强行占用某个端口的属性 */
	nValue = 1;
	if (setsockopt(nSocket, SOL_SOCKET, SO_REUSEADDR, &nValue, sizeof(int)) == -1)
	{
		//NetError(errStr, "Error on setsockopt SO_REUSEADDR: ", errno);
		close(nSocket);
		return -2;
	}

	if (-1 == AnetKeepAlive( nSocket))
	{
		close(nSocket);
		return -3;
	}

	struct sockaddr_in sin;
	memset(&sin, 0, sizeof(sin));

	sin.sin_family = AF_INET;
	sin.sin_addr.s_addr = htonl(INADDR_ANY);/* 本机所有网卡 */
	sin.sin_port = htons(port);

	int ret = bind(nSocket, (struct sockaddr*)&sin, sizeof(sin));
	if (-1 == ret)
	{
		//NetError(errStr, "Error on bind socket: ", errno);
		return -4;
	}

	if (-1 == listen(nSocket, 1024))
	{
		//NetError(errStr, "Error on listen socket : ", errno);
		return -5;
	}

	return nSocket;
}
int SLTcpServerEpollImpl::AnetKeepAlive(int fd)
{
	/*
		keepIdle        如该连接在6秒内没有任何数据往来,则进行探测
		keepInterval    探测时发包的时间间隔
		keepCount = 5   探测尝试的次数.如果第1次探测包就收到响应了,则后4次的不再发.
	*/

	int nKeepIdle = 6, nKeepInterval = 1, nKeepCount = 5;
	int val = 1;

	if (setsockopt(fd, SOL_SOCKET, SO_KEEPALIVE, &val, sizeof(int)) == -1)
	{
		//NetError(errStr, "Error setsockopt SO_KEEPALIVE: ", errno);
		return -1;
	}

	if (setsockopt(fd, IPPROTO_TCP, TCP_KEEPIDLE, &nKeepIdle, sizeof(nKeepIdle)) < 0)
	{
		//NetError(errStr, "Error setsockopt TCP_KEEPIDLE :", errno);
		return -2;
	}

	if (setsockopt(fd, IPPROTO_TCP, TCP_KEEPCNT, &nKeepCount, sizeof(nKeepCount)) < 0)
	{
		//NetError(errStr, "Error setsockopt TCP_KEEPCNT : ", errno);
		return -3;
	}

	if (setsockopt(fd, IPPROTO_TCP, TCP_KEEPINTVL, &nKeepInterval, sizeof(nKeepInterval)) < 0)
	{
		//NetError(errStr, "Error setsockopt TCP_KEEPINTVL : ", errno);
		return -4;
	}
	return 0;
}
void SLTcpServerEpollImpl::NetClose(int socket)
{
	if (socket == -1) {
		return;
	}
	close(socket);
}
SLTcpClientPtr  SLTcpServerEpollImpl::createClient()
{
	SLTcpClientEpollImplPtr pc = std::make_shared<SLTcpClientEpollImpl>((uint32_t)m_clientIDBase++);
	pc->m_pserver = this;
	return std::static_pointer_cast<SLTcpClient>(pc);
	//return nullptr;
}
#define TcpServer_EPOLL_MAX_CACHE  500
void SLTcpServerEpollImpl::thread_listen(void)
{
	std::string client;
	char errStr[1024] = { 0 };

	m_epoll_fd = epoll_create(TcpServer_EPOLL_MAX_CACHE);
	if (-1 == m_epoll_fd) {
		_TCPSERVER_LOG_ERR("epoll_create error, info : %d", errno);
		return;
	}

	struct epoll_event _events[TcpServer_EPOLL_MAX_CACHE], ev;
	ev.events = EPOLLET | EPOLLHUP | EPOLLIN | EPOLLRDHUP | EPOLLERR;
	ev.data.fd = m_QuitEventfd;
	EpollAdd(m_QuitEventfd, &ev);

	int _nfds_num = 0, nSocket;
	while (1)
	{
		_nfds_num = epoll_wait(m_epoll_fd, _events, TcpServer_EPOLL_MAX_CACHE, m_accpetTimeout); /* 没有活跃句柄，持续等待 */
		if (-1 == _nfds_num)
		{
			if (errno == EINTR) {
				continue;
			}
			_TCPSERVER_LOG_ERR("epoll wait error , %d", errno);
			break;
		}
		if(m_accpetTimeout > 0){
			m_accpetTimeout = -1;
			auto lIter = m_accpentList.begin();
			for(;lIter != m_accpentList.end();){
				if(recvConnect(m_accpentList.front()) == 0){
					lIter = m_accpentList.erase(lIter);
				}else{
					m_accpetTimeout = 100;
					++lIter;
				}
			}
		}

		for (int index = 0; index < _nfds_num; ++index)
		{
			//printf("[tcpser]event, info:%d, fd:%d,event:%X\n", errno, _events[index].data.fd, _events[index].events);
			if (_events[index].data.fd == m_QuitEventfd) {
				HandlingEvent(m_QuitEventfd);
			}
			else {
				auto sockIter = m_clientMap.find(_events[index].data.fd);
				if (sockIter != m_clientMap.end()) {
					if (_events[index].events & EPOLLOUT) {
						//todo  当EPOLLIN | EPOLLOUT 同时使用时，EPOLLIN触发时会同时触发EPOLLOUT.有些浪费，先这样吧
						//_TCPSERVER_LOG_ERR("epoll out event fd:%d", _events[index].data.fd);
						sockIter->second->onSend2Net();
					}

					if (_events[index].events & EPOLLIN) {
						if (sockIter->second->onRecvFromNet() < 0) {
							_events[index].events = 0/*EPOLLIN | EPOLLOUT*/;   //清除关闭事件，避免没有接收数据的情况下，远端关闭了。上层保证逻辑
						}

						if (_events[index].events & EPOLLRDHUP || _events[index].events & EPOLLERR || _events[index].events & EPOLLHUP)
						{
							_TCPSERVER_LOG_ERR("exception close,  fd:%d,events:%X,errno:%d",  _events[index].data.fd, _events[index].events, errno);
							sockIter->second->onError(tcp_error_type_err);
						}
					}
						
				}
				else {
					auto listenIter = m_listenMap.find(_events[index].data.fd);
					if (listenIter != m_listenMap.end()) {
						if (_events[index].events & EPOLLIN) {
							if(recvConnect(listenIter->second)< 0){
								m_accpetTimeout = 100;  //100ms
								m_accpentList.push_back(listenIter->second);
							}
						}
						if (_events[index].events & EPOLLRDHUP || _events[index].events & EPOLLERR || _events[index].events & EPOLLHUP)
						{
							_TCPSERVER_LOG_ERR("exception close，listen fd:%d,events:%X,errno:%d", _events[index].data.fd, _events[index].events, errno);
							m_listenMap.erase(_events[index].data.fd);

							EpollDel(_events[index].data.fd);
							NetClose(_events[index].data.fd);
						}
					}
				}
			}
	
		}
	}
}
int SLTcpServerEpollImpl::recvConnect(_tcp_listen_info& _listenInfo)
{
	struct epoll_event ev;
	int rip;
	int rport;
	while (1) {
		int nSocket = NetAcceptSocket(_listenInfo.m_listenSocket, rip, rport);
		if (-1 == nSocket) {
			_TCPSERVER_LOG_ERR("[tcpser]accept error:fail, ret:%d", nSocket);
			return -1;
		}
		else if (-2 == nSocket) {
			//printf("[tcpser]accept error:again, ret2:%d\n", nSocket);
			return 0;
		}
		//SLTcpClient::NoBlock(nSocket);
		evutil_make_socket_nonblocking(nSocket);

		SLTcpClientEpollImplPtr pnewClient = std::make_shared<SLTcpClientEpollImpl>(m_clientIDBase++);
		if (pnewClient) {
			ev.data.fd = nSocket;
			ev.events = EPOLLET | EPOLLHUP | EPOLLIN | EPOLLRDHUP | EPOLLERR| EPOLLOUT;
			EpollAdd(nSocket, &ev);

			pnewClient->m_lip = _listenInfo.m_ip;
			pnewClient->m_lport = _listenInfo.m_port;
			pnewClient->m_socket = nSocket;
			pnewClient->m_rip = rip;
			pnewClient->m_rport = rport;
			pnewClient->m_pserver = this;

			_TCPSERVER_LOG_INFO("accept,%s", pnewClient->toString().c_str());
			pnewClient->m_workState = SLTcpClientEpollImpl::tcp_work_state_ok;
			m_clientMap.insert(make_pair(nSocket, pnewClient));
			SLTcpClientPtr tptr = std::static_pointer_cast<SLTcpClient>(pnewClient);
			m_pcb->TcpServer_onConnect(tptr);
		}
		else {
			_TCPSERVER_LOG_ERR("new connect,memory error!!.lport:%d", _listenInfo.m_port);
			NetClose(nSocket);
		}
	}
}
void SLTcpServerEpollImpl::postEvent(_tcp_event_info& _ev)
{
	m_eventInfoListLock.lock();
	m_eventInfoList.push_back(_ev);
	if (m_eventInfoList.size() == 1) {
		uint64_t flag = 1;
		if (sizeof(uint64_t) != write(m_QuitEventfd, &flag, sizeof(uint64_t))) {
			_TCPSERVER_LOG_ERR("Error on SendEventSig : %d", errno);
		}
	}
	m_eventInfoListLock.unlock();
}
void SLTcpServerEpollImpl::postEvent2(uint32_t cmd, uint32_t fd)
{
	_tcp_event_info ev;
	ev.m_cmd = cmd;
	ev.m_socket = fd;
	postEvent(ev);
}
//int SLTcpServerEpollImpl::SendEventSig(uint32_t data, uint32_t cmd)
//{
//	uint64_t flag = ((uint64_t)data) << 32 | cmd;
//	if (sizeof(uint64_t) != write(m_QuitEventfd, &flag, sizeof(uint64_t))) {
//		_TCPSERVER_LOG_ERR("Error on SendEventSig : %d", errno);
//		return -1;
//	}
//	return 0;
//}
int SLTcpServerEpollImpl::HandlingEvent(int eventFd)
{
	uint64_t eventData = 0;
	if (sizeof(uint64_t) != read(eventFd, &eventData, sizeof(uint64_t)))
	{
		_TCPSERVER_LOG_ERR("read event error, %d", errno);
	}

	HandlingEventList();
#if 0
	int cmd = eventData & 0xFFFFFFFF;
	int nSocket = eventData >> 32;
	SLT_LOG_DEBUG("[tcpser]recv event, cmd : %u, value : %u\n", cmd, nSocket);
	switch (cmd)
	{
	case SHUTDOWNFLG: {
		break;
	}
					//case EPOLL_SOCKET_RECV:{
					//	//通过connect，accpent 已经添加到map，这里只修改epoll状态
					//	struct epoll_event ev;
					//	ev.data.fd = nSocket;
					//	ev.events = EPOLLET | EPOLLHUP | EPOLLIN | EPOLLRDHUP | EPOLLERR;
					//	EpollAdd(nSocket, &ev);
					//	break;
					//}
					/*case EPOLL_SOCKET_DEL: {
						auto iter = m_clientMap.find(nSocket);
						if (iter != m_clientMap.end())
						{
							iter->second->onError(tcp_error_type_localclose);
							m_clientMap.erase(iter);

							EpollDel(nSocket);
							NetWork::NetClose(nSocket);
						}
						break;
					}*/
					/*case EPOLL_SOCKET_RECVDATA: {
						auto iter = m_clientMap.find(nSocket);
						if (iter != m_clientMap.end())
						{
							recvData(iter->second);
						}
						break;
					}*/
	case EPOLL_LIST_CMD: {
		HandlingEventList();
		break;
	}
	default: {
		printf("[tcpser] invalid cmd : %d\n", cmd);
		break;
	}
	}
#endif
	return 0;
}
void SLTcpServerEpollImpl::HandlingEventList()
{
	_tcp_event_info evinfo;
	while (1) {
		m_eventInfoListLock.lock();
		if (m_eventInfoList.empty()) {
			m_eventInfoListLock.unlock();
			return;
		}
		evinfo = m_eventInfoList.front();
		m_eventInfoList.pop_front();
		m_eventInfoListLock.unlock();

		if (EPOLL_LIST_SOCKET_CHECKTIMEOUT != evinfo.m_cmd) {
			_TCPSERVER_LOG_DEBUG("event,cmd:%u,fd:%d", evinfo.m_cmd, evinfo.m_socket);
		}

		switch (evinfo.m_cmd)
		{
		case EPOLL_LIST_CMD_ADD_LISTEN_SOCKET: {
			struct epoll_event ev;
			ev.data.fd = evinfo.m_socket;
			ev.events = EPOLLET | EPOLLHUP | EPOLLIN | EPOLLRDHUP | EPOLLERR;
			EpollAdd(evinfo.m_socket, &ev);

			_tcp_listen_info linfo;
			linfo.m_listenSocket = evinfo.m_socket;
			linfo.m_port = evinfo.m_port;
			m_listenMap.insert(std::make_pair(evinfo.m_socket, linfo));
			break;
		}
		case EPOLL_LIST_ADD_CONNECT: {
			m_clientMap.insert(std::make_pair(evinfo.m_socket, std::static_pointer_cast<SLTcpClientEpollImpl>(evinfo.m_clientPtr)));
			struct epoll_event ev;
			ev.data.fd = evinfo.m_socket;
			ev.events = EPOLLET | EPOLLHUP | EPOLLIN | EPOLLRDHUP | EPOLLERR| EPOLLOUT;
			EpollAdd(evinfo.m_socket, &ev);
			break;
		}
		case EPOLL_LIST_SOCKET_DEL: {
			auto iter = m_clientMap.find(evinfo.m_socket);
			if (iter != m_clientMap.end())
			{
				iter->second->onError(tcp_error_type_localclose);
				EpollDel(evinfo.m_socket);

				m_clientMap.erase(iter);
				NetClose(evinfo.m_socket);
			}
			break;
		}
		case EPOLL_LIST_SOCKET_STARTRECV:
		case EPOLL_LIST_SOCKET_RECVDATAONE: {
			auto iter = m_clientMap.find(evinfo.m_socket);
			if (iter != m_clientMap.end())
			{
				iter->second->onRecvFromNet();
				//recvData(iter->second);
			}
			break;
		}
		case EPOLL_LIST_SOCKET_CHECKTIMEOUT: {
			checkTimeOut();
			break;
		}
		default:
			_TCPSERVER_LOG_ERR("event: unkown cmd:%d,fd:%d", evinfo.m_cmd, evinfo.m_socket);
		}
	}
}

int SLTcpServerEpollImpl::NetAcceptSocket(int socket, int& _rip, int& _rport)
{
	struct sockaddr_in cli_addr;
	int nAddrSize = sizeof(cli_addr);
	memset(&cli_addr, 0, sizeof(cli_addr));
	int nAccept;
	while (1) {
		nAccept = accept(socket, (struct sockaddr*)&cli_addr, (socklen_t*)&nAddrSize);
		if (-1 == nAccept) {
			if (EINTR == errno) {
				continue;
			}
			else if (errno == EWOULDBLOCK || errno == EAGAIN) {
				return -2;
			}
			else {
				_TCPSERVER_LOG_ERR("Error on accept,%d\n", errno);
				return -1;
			}
		}
		break;
	}

	_rip = ntohl(cli_addr.sin_addr.s_addr);
	_rport = ntohs(cli_addr.sin_port);
	return nAccept;
}
void SLTcpServerEpollImpl::EpollDel(int nSocket)
{
	if (0 != epoll_ctl(m_epoll_fd, EPOLL_CTL_DEL, nSocket, NULL))
	{
		_TCPSERVER_LOG_ERR("EPOLL_CTL_DEL,error ,info : %s", strerror(errno));
	}
}
void SLTcpServerEpollImpl::EpollMod(int nSocket, struct epoll_event* ev)
{
	if (0 != epoll_ctl(m_epoll_fd, EPOLL_CTL_MOD, nSocket, ev))
	{
		_TCPSERVER_LOG_ERR("EPOLL_CTL_MOD, info : %s", strerror(errno));
	}
}
void SLTcpServerEpollImpl::EpollAdd(int nSocket, struct epoll_event* ev)
{
	if (0 != epoll_ctl(m_epoll_fd, EPOLL_CTL_ADD, nSocket, ev))
	{
		_TCPSERVER_LOG_ERR("EPOLL_CTL_ADD, info : %s", strerror(errno));
	}
}
void SLTcpServerEpollImpl::checkTimeOut()
{
	for (auto& cli : m_clientMap) {
		if (cli.second->m_isActive) {
			cli.second->m_isActive = false;
			cli.second->m_lastActiveTime.start();
		}
		else {
			//超过3分钟
			if (cli.second->m_lastActiveTime.elapseMs() > 180000) {
				cli.second->onError(tcp_error_type_timeout);
			}
		}
	}
}
/******************************************************************************************************************/
void SLTcpTimeoutCheckSession::OnStart()
{
	setTimer(1, 30000, false);
}
void SLTcpTimeoutCheckSession::OnTimer(uint32_t timerID, int _intervalMs, bool periodic)
{
	setTimer(1, 30000, false);
	_tcp_event_info ev;
	ev.m_cmd = EPOLL_LIST_SOCKET_CHECKTIMEOUT;
	m_ptcpServer->postEvent(ev);
}
